package com.xiaoniu.mc.client.enums;

import com.xiaoniu.mc.client.handler.XmqMessageHook;
import com.xiaoniu.mc.client.message.RecieveMessageWrapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.ArrayList;
import java.util.List;

/**
 * @类描述：
 * @创建人：林继丰
 * @创建时间：2017/8/17 17:36
 * @版权：Copyright (c) 深圳市牛鼎丰科技有限公司-版权所有.
 */
public enum ConsumeStrategy {

    PUSH_MODEL_CONCURRENCY {
        @Override
        public void registerHooks(DefaultMQPushConsumer mqPushConsumer, final XmqMessageHook topicMapHooks) {
            MessageListenerConcurrently listenerConcurrently = new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    boolean isAllConsume = dispatcherMessage(list, topicMapHooks);
                    if (isAllConsume)
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    else
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            };
            mqPushConsumer.setMessageListener(listenerConcurrently);
        }
    },

    PUSH_MODEL_ORDERLY {
        @Override
        public void registerHooks(DefaultMQPushConsumer mqPushConsumer, final XmqMessageHook topicMapHooks) {
            MessageListenerOrderly listener = new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    boolean isAllConsume = dispatcherMessage(list, topicMapHooks);
                    if (isAllConsume)
                        return ConsumeOrderlyStatus.SUCCESS;
                    else
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            };
            mqPushConsumer.setMessageListener(listener);

        }
    },

    PULL_MODE_CONCURRENCY {},

    PULL_MODE_ORDERLY {};

    ConsumeStrategy() {

    }

    public void registerHooks(DefaultMQPushConsumer mqPushConsumer, XmqMessageHook topicMapHooks) {
    }

    protected boolean dispatcherMessage(List<MessageExt> msgs, XmqMessageHook hook) {
        if (msgs.size() == 0)
            return false;
        boolean isAllConsume = true;
        List<RecieveMessageWrapper> prepareToSendMessages = new ArrayList<RecieveMessageWrapper>(msgs.size());
        for (MessageExt msg : msgs) {
            RecieveMessageWrapper wrapper = new RecieveMessageWrapper(msg.getBody(), msg.getQueueId(), msg.getTopic(), msg.getTags(), msg.getKeys());
            prepareToSendMessages.add(wrapper);
        }
        try {
            hook.handle(prepareToSendMessages);
        } catch (Exception e) {
            e.printStackTrace();
            isAllConsume = false;
        }
        //TODO 批量处理的时候，业务自己做幂等处理
        return isAllConsume;
    }
}
